package org.wisdom.mq.kafka.service;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;

/**
 * 消费消息
 *
 * @author 张智研
 * @date 2023/11/09 09:33
 **/
@Component
@Slf4j
public class KafkaConsumer {
    private final String topicName = "sb_topic";
    @KafkaListener(
            topics = {topicName},
            groupId = "test-group")
    public void consumeMsg(String record) {
        try {
            log.info("从Kafka中消费的原始数据: " + record);
        } catch (Exception e) {
            log.info("Consumer消费Topic异常.");
            e.printStackTrace();
        }
    }

    @KafkaListener(
            topics = {"flume_topic"},
            groupId = "flume-consumer-group")
    public void consumeFlumeMsg(String record) {
        try {
            log.info(">> " + record);
        } catch (Exception e) {
            log.info("Consumer消费Topic异常.");
            e.printStackTrace();
        }
    }
}
